package cn.iotab.iotkit.transport.mqtt;

import cn.iotab.iotkit.transport.api.driver.TransportDriver;
import cn.iotab.iotkit.transport.api.driver.TransportDriverConfig;
import cn.iotab.iotkit.transport.api.driver.TransportDriverService;
import cn.iotab.iotkit.transport.mqtt.conf.MqttServerConfig;
import cn.iotab.iotkit.transport.mqtt.ssl.MqttSslHandlerProvider;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.ResourceLeakDetector;
import lombok.extern.slf4j.Slf4j;


/**
 * @author Andrew Shvayka
 */
@Slf4j
public class MqttTransportServer {

    private final TransportDriverConfig transportDriverConfig;

    private final MqttServerConfig config;

    public MqttTransportServer(MqttServerConfig config, TransportDriverConfig transportDriverConfig) {
        this.config = config;
        this.transportDriverConfig = transportDriverConfig;
    }

    private TransportDriver transportdriver;

    private Channel serverChannel;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    public void init(){
        transportdriver = new TransportDriver(transportDriverConfig);
        transportdriver.init();
    }


    public void start() throws Exception {
        TransportDriverService transportDriverService = transportdriver.getTransportDriverService();
        MqttSslHandlerProvider sslHandlerProvider = new MqttSslHandlerProvider(config.getSslConf(), transportDriverService);
        //new transport context
        MqttTransportContext context = MqttTransportContext.builder()
                .transportDriverService(transportDriverService)
                .sslHandlerProvider(sslHandlerProvider)
                .skipValidityCheckForClientCert(config.getSslConf().isSkipValidityCheckForClientCert()).build();
        //
        doStartMqttServer(context);
    }

    private void doStartMqttServer(MqttTransportContext transportContext) throws Exception {

        MqttServerConfig.NettyConf nettyConf = config.getNettyConf();

        log.info("Setting resource leak detector level to {}", nettyConf.getLeakDetectorLevel());
        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(nettyConf.getLeakDetectorLevel().toUpperCase()));

        log.info("Starting MQTT transport...");
        bossGroup = new NioEventLoopGroup(nettyConf.getBossGroupThreadCount());
        workerGroup = new NioEventLoopGroup(nettyConf.getWorkerGroupThreadCount());
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new MqttTransportServerInitializer(config, transportContext))
                .childOption(ChannelOption.SO_KEEPALIVE, nettyConf.isKeepAlive());

        serverChannel = b.bind(config.getHost(), config.getPort()).sync().channel();
        log.info("Mqtt transport started!");

    }

    private void doShutdownMqttServer() throws InterruptedException {
        log.info("Stopping MQTT transport!");
        try {
            serverChannel.close().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
        log.info("MQTT transport stopped!");
    }

    public void destroy(){
        try {
            doShutdownMqttServer();
        } catch (InterruptedException e) {
            log.warn("shutdown mqtt server InterruptedException");
        }
        if (transportdriver != null){
            transportdriver.destroy();
        }
    }
}
